package com.jenvin.activemq.topic;

import java.util.Arrays;

import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.jms.TopicSubscriber;

import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * Use in conjunction with TopicPublisher to test the performance of ActiveMQ Topics.
 */
public class TopicListener implements MessageListener {

    private String  url       = "tcp://192.168.1.62:61616";
    private Boolean isAck     = true;
    private String  clientId  = "connection1";
    private String  topicName = "topic.test.ask2";

    public static void main(String[] argv) throws Exception {
        TopicListener l = new TopicListener();
        String[] unknown = CommandLineSupport.setOptions(l, argv);
        if (unknown.length > 0) {
            System.out.println("Unknown options: " + Arrays.toString(unknown));
            System.exit(-1);
        }
        l.run();
    }

    public void run() throws JMSException {
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(url);
        Connection connection;
        Session session;
        Topic topic;
        connection = factory.createConnection();
        connection.setClientID(clientId);
        session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
        topic = session.createTopic(topicName);
        TopicSubscriber topicSubscriber = session.createDurableSubscriber(topic, "name");
        topicSubscriber.setMessageListener(this);
        connection.start();
        System.out.println("exit");
    }

    public void onMessage(Message message) {
        TextMessage msg = (TextMessage) message;
        try {
            System.out.println(msg.getText());
            if (isAck) message.acknowledge();
            Thread.sleep(1000);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void setUrl(String url) {
        this.url = url;
    }

    public Boolean getIsAck() {
        return isAck;
    }

    public void setIsAck(Boolean isAck) {
        this.isAck = isAck;
    }

    public String getClientId() {
        return clientId;
    }

    public void setClientId(String clientId) {
        this.clientId = clientId;
    }

    public String getTopicName() {
        return topicName;
    }

    public void setTopicName(String topicName) {
        this.topicName = topicName;
    }

    public String getUrl() {
        return url;
    }

}
